package org.company.mars.kafka;

import java.util.Properties;

import org.company.earth.util.JsonUtil;

import com.fasterxml.jackson.databind.node.ObjectNode;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class KafkaProducer {
	
	public void send(String message) throws InterruptedException {
		Properties props = new Properties();
		props.setProperty("metadata.broker.list", "l-kafka1.f.dev.cn0.qunar.com:9092,l-kafka2.f.dev.cn0.qunar.com:9092,l-kafka3.f.dev.cn0.qunar.com:9092");
		//props.setProperty("metadata.broker.list", "localhost:9092");
		props.put("serializer.class", "kafka.serializer.StringEncoder");
		//props.put("partitioner.class", "example.producer.SimplePartitioner");
		props.put("request.required.acks", "1");
		
		ProducerConfig config = new ProducerConfig(props);
		
		Producer<String, String> producer = new Producer<String,String>(config);
		ObjectNode node = null;
		try {
			for(int i = 0 ; i < 100 ; i++) {
				node =  JsonUtil.getInstance().createNode();
				node.put("source_path", "/home/q/www/ich.qunar.com/logs/web.log");
				node.put("source_host", "localhost");
				node.put("source_ip", "127.0.0.1");
				node.put("category", "logs.logger.sp");
				node.put("timestamp", System.currentTimeMillis());
				node.put("app_code", "sp_ich");
				node.put("send_time", System.currentTimeMillis());
				node.put("content", message);
				KeyedMessage<String, String> data = new KeyedMessage<String, String>("logs.logger.sp",node.toString());
				producer.send(data);
			}
		} finally {
			producer.close();
		}
	}
	
	public static void main(String[] args) throws InterruptedException {
		KafkaProducer producer = new KafkaProducer();
		producer.send("2015-11-14 20:06:16,708 [http-bio-8080-exec-9] INFO  com.ich.logic.util.WebLog.transform2DTO[1035] - v=V3.08&di=f321b587&ld=0&gi=84fc2e71&p=c1&ss=72&ct=lh");
	}
}
